package xx.rocketmq.producer;

import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
@Slf4j
@Component
public class SendMsg {

    @Resource
    RocketMQTemplate rocketMQTemplate;


    // 统一用一个TOPIC
    private static final String SIMPLE_TOPIC = "SIMPLE_TOPIC";


    /**
     * 发送普通的消息
     *
     * @param msg
     */
    public void send(String msg) {
        rocketMQTemplate.convertAndSend(SIMPLE_TOPIC, msg);
    }

    /**
     * 发送同步消息 （会阻塞当前线程，等待broker响应发送结果，这样不太容易丢失数据）
     * @param msg
     */
    public void syncSend(String msg){
        SendResult sendResult =  rocketMQTemplate.syncSend(SIMPLE_TOPIC, msg);
        log.info("发送同步消息结果：", JSON.toJSONString(sendResult));
    }

    /**
     * 发送异步消息（通过线程池执行发送到broker的消息任务，执行完后回调，在sendCallback中可处理相关成功和失败的逻辑）
     * 适合 对响应时间敏感的业务场景
     * @param msg
     */
    public void sendAsyncMsg(String msg){
        rocketMQTemplate.asyncSend(SIMPLE_TOPIC, MessageBuilder.withPayload(msg).build(), new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("异步发送消息成功");
            }

            @Override
            public void onException(Throwable throwable) {
                log.info("异步发送消息异常啦");
            }
        });
    }


    /**
     * 发送延时消息（上面发送的是同步消息，delayLevel 的值为0，因为不延时）
     * 在start 版本中，延时消息一共分为： 1s ,5s ,10s ,30s 1m 2m 3m 4m 5m - 10m 20m 30m 1h 2h
     * @param msg
     * @param delayLevel
     */
    public void sendDelayMsg(String msg,int delayLevel){
        rocketMQTemplate.syncSend(SIMPLE_TOPIC, MessageBuilder.withPayload(msg).build(),3000,delayLevel);
    }


    /**
     * 发送单向消息（只负责发送消息，不等待应答，不关心发送结果）
     * @param msg
     */
    public void sendOneMsg(String msg){
        rocketMQTemplate.sendOneWay(SIMPLE_TOPIC,MessageBuilder.withPayload(msg).build());
    }

    /**
     * 发送带 tag的消息 直接在topic后面加上 :tag即可
     * @param msg
     * @return
     */
    public SendResult sendTagMsg(String msg){
        return rocketMQTemplate.syncSend(SIMPLE_TOPIC+":tag2",MessageBuilder.withPayload(msg).build());
    }
}
